package rabbitmq.test;

import com.rabbitmq.client.*;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * 专门消费监听的测试类
 */
public class MyConsumer {
    private Channel channel;
    @Before
    public void init(){
        //1.创建长连接 提供属性(ip,port,username,password,virtualHost)
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("192.168.8.100");
        factory.setPort(5672);//TCP协议端口
        factory.setUsername("admin");
        factory.setPassword("123456");
        factory.setVirtualHost("/");
        try{
            Connection connection = factory.newConnection();
            //2.从长连接赋值短链接
            channel=connection.createChannel();
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("创建连接失败");
        }
    }
    //测试消费逻辑代码
    @Test
    public void receiveMessage01() throws Exception{
        //监听queue01测试队列 获取其中的消息数据调用channel 的方法
        /**
         * 参数解释
         * String queue: 消费者监听绑定的队列 一个basiceConsume方法可以监听一个队列
         * Boolean autoAck: 是否自动确认 true每个消息达到消费者之后 立刻反馈给队列消费成功 false需要手动确认
         * DeliverCallback: 回调(异步线程调用的程序代码),处理消费逻辑的方法
         * CancelCallback: 消费取消回调,监听取消消费的回调方法
         */
        DeliverCallback deliver=new DeliverCallback() {
            //每当队列传递一条消息到达消费者进行消费的是偶 handle就会被调用一次
            //所以这个deliver的handle方法就是消费者处理消费逻辑核心代码

            /**
             * 参数解释
             * String consumerTag: 消费者标签,用来区分多个消费者,每个消费者有个唯一的标签
             * Delivery message: rabbitmq投递过来的消息数据
             */
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                //解析message和生产发送的参数是对应的
                message.getEnvelope().getExchange();
                message.getEnvelope().getRoutingKey();
                AMQP.BasicProperties properties = message.getProperties();
                String messageId = properties.getMessageId();//消息id 默认生成的
                System.out.println("本次消息的id:"+messageId);
                //解析消息内容 反序列化
                byte[] body = message.getBody();
                String msg=new String(body, StandardCharsets.UTF_8);
                System.out.println("消费到的消息:"+msg);
                //如果发现某一条消息 触发了临界取消逻辑 只允许该消费者消费1000条数据
                //channel.basicCancel(consumerTag);
                //代码已经消费完毕 应该反馈确认
                //channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
                channel.basicReject(message.getEnvelope().getDeliveryTag(),false);
            }
        };
        CancelCallback cancel=new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {
                //当消费者取消消费的时候 cancel会调用 channel.basicCancel();
                System.out.println("消费者取消消费");
            }
        };
        channel.basicConsume(
                "ex_queue05",
                false,
                deliver,
                cancel);
        while(true);
    }
}
